package com.alex.quickstart;

import com.alex.constants.Const;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;

/**
 * DefaultMQPushConsumer
 *  1. 采用长轮训机制拉取数据；broker会hold住一些consumer线程，所以，对于consumer的个数要可控，防止资源溢出
 */
@Slf4j
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.设置所属组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_start_consumer");
        //2.设置nameServer
        consumer.setNamesrvAddr(Const.NAMESRV_ADDR);
        //3.consumer的offset从哪开始消费;
        // 只针对第一次程序启动时设定; 后续程序再次启动, 会读取broker的offset进行消费.
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //4.订阅topic和tag
        /**
         * Tag：在历史版本中, Tag支持‘TagA | TagB’这种运算符；但是4.x之后，就不允许这样的tag配置
         * 原因：提升性能；这个Tag的过滤放到了broker中进行；consumer去broker拉取消息的时候，只能拉取到对应Tag的消息。
         */
        consumer.subscribe("test-quick-topic", "*");

        //队列策略; 默认为avg.
        //consumer.setAllocateMessageQueueStrategy();

        // 指定本地存储offset或者远程存储offset
        //consumer.setOffsetStore();

        //提高consumer并行消费能力
        //consumer.setConsumeThreadMin();
        //consumer.setConsumeThreadMax();

        // 流控1： 单个队列并行消费的最大跨度
        //consumer.setConsumeConcurrentlyMaxSpan();

        // 流控2：单个队列最大的消息个数
        //consumer.setPullThresholdForQueue();

        // 消息拉取时间间隔(用于PullConsumer)
        //consumer.setPullInterval();

        // 一次性拉取的消息个数(用于PullConsumer)
        //consumer.setPullBatchSize();

        // 一次可以拉取多少条数据，default：1
        //consumer.setConsumeMessageBatchMaxSize();

        //最大重试次数
        //consumer.setMaxReconsumeTimes();

        // 消费模式
        // CLUSTERING: 集群，每个消费组，消费部分消息(Default)
            // GroupName用于把多个Consumer组织到一起
            // Consumer监听的是一个Topic下的多个Queue;
            // 完全的负载均衡：保证consumer是queueNum的倍数
        // BROADCASTING: 广播，每个消费组，消费全部消息
        //consumer.setMessageModel(MessageModel.CLUSTERING);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgGroup, context) -> {
            for (MessageExt msg : msgGroup) {
                try {

                    String topic = msg.getTopic();
                    String tags = msg.getTags();
                    String keys = msg.getKeys();

                    if (keys.equals("key1")) {
                        // 模拟consumer发生异常, 触发rocketMq重发机制
                        int a = 1 / 0;
                    }
                    String msgBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    log.info("topic: {}, tags:{}, keys:{}, body:{}, queuId: {}", topic, tags, keys, msgBody
                    , msg.getQueueId());
                } catch (Exception ex) {
                    log.error("Consumer Exception", ex);

                    // 这样的操作是防止, 错误的消息不断重发; 而用户无法感知到错误信息的发生.
                    int reconsumeTimes = msg.getReconsumeTimes();
                    if (reconsumeTimes == 3) {
                        // 记录日志
                        // 补偿处理异常消息
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    // 过一段时间再次重发
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();

    }
}
